package com.haozhen.pgproject

import org.apache.spark.sql.ForeachWriter

/**
  * @author haozhen
  * @email haozh@ync1.com
  * @date 2021/2/7  11:50
  */
class JdbcWriter extends ForeachWriter[BusInfo]{

  import java.sql.{Connection, PreparedStatement}

  var conn :Connection=_

  var statement: PreparedStatement = _



  override def open(partitionId: Long, epochId: Long): Boolean = {
    if (conn == null){
      conn = JdbcWriter.openConnection
    }
    true
  }

  override def process(value: BusInfo): Unit = {
    val arr  = value.lglat.split("_")
    val sql = "insert into car_gps(deployNum,plateNum,timeStr,lng,lat)" +
      "value(?,?,?,?,?)"
    statement = conn.prepareStatement(sql)
    statement.setString(1,value.deployNum)
    statement.setString(2,value.plateNum)
    statement.setString(3,value.timeStr)
    statement.setString(4,arr(0))
    statement.setString(5,arr(1))
    statement.executeUpdate()
  }

  override def close(errorOrNull: Throwable): Unit = {
    if(null!=statement) statement.close()
    if(null!=conn) conn.close()
  }
}

object JdbcWriter {

  import java.sql.Connection

  var conn:Connection=_

  val url = "jdbc:mysql://test1:3306/gp_app1?useUnicode=true&characterEncoding=utf8"

  val username="root"
  val password="123456"
  def openConnection:Connection={
  if(null==conn || conn.isClosed){
    import java.sql.DriverManager
    import java.util.Properties
    val p = new Properties
    Class.forName("com.mysql.jdbc.Driver")
    conn = DriverManager.getConnection(url,username,password)
  }
  conn
  }
}
